
package org.halk.job_admin.config;

import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.lang.UUID;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.halk.job_admin.mapper.ScheduleLogMapper;
import org.halk.job_admin.mapper.ScheduleMapper;
import org.halk.job_admin.mq.produce.JobSendProduce;
import org.halk.job_admin.pojo.entity.JobRequest;
import org.halk.job_admin.pojo.entity.Schedule;
import org.halk.job_admin.pojo.entity.ScheduleLog;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTask;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.config.TriggerTask;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 注册定时任务
 * 先注册守护定时任务
 * 由守护定时任务维护业务任务的状态、心跳，是否删除、是否重新加载
 * 如果不用额外的守护定时任务，
 *
 * @Author haozhu7
 * @Date 2024/9/20 16:08
 **/
@Component
@Slf4j
public class JobScheduleTask implements SchedulingConfigurer {

    @Resource
    private ScheduleMapper scheduleMapper;
    @Resource
    private ScheduleLogMapper scheduleLogMapper;
    @Resource
    private JobSendProduce jobSendProduce;

    private static final String DAEMON = "daemon";
    private static final String BUSINESS = "business";

    private static final Map<String, ScheduledTask> runTaskMap = new HashMap<>();

    /**
     * 此方法只会在springboot启动时加载一次，
     * 装载守护定时任务，守护定时任务会定时检查业务定时任务，如果业务定时任务有更新，则重新加载
     *
     * @param taskRegistrar
     */
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        log.info("定时任务配置开始");
        loadDaemon(taskRegistrar);
        log.info("定时任务配置完成");
    }

    private void loadDaemon(ScheduledTaskRegistrar taskRegistrar) {
        List<Schedule> scheduleList = this.scheduleMapper.selectList(new QueryWrapper<Schedule>()
                .eq("enabled", 1)
                .eq("type", "daemon")
        );
        for (Schedule schedule : scheduleList) {
            taskRegistrar.addTriggerTask(newTask(taskRegistrar, schedule), getTrigger(schedule));
        }
    }

    private Runnable newTask(ScheduledTaskRegistrar taskRegistrar, Schedule schedule) {
        if (DAEMON.equals(schedule.getType()) && schedule.getEnabled() == 1) {
            return newDaemonTask(taskRegistrar);
        } else if (BUSINESS.equals(schedule.getType()) && schedule.getEnabled() == 1) {
            return newBusinessTask(schedule);
        } else {
            log.warn("[job] 定时器任务 '{}' 类型 '{}' 暂不支持。 ", schedule.getScheduleId(), schedule.getType());
            return () -> {
            };
        }
    }

    private Runnable newBusinessTask(Schedule schedule) {
        return () -> {
            //判断任务是否可以运行，同步任务需要等上一个任务结束才能运行，异步不需要
            if (!judgeRun(schedule)) {
                return;
            }

            //将任务信息发送到mq中，由mq消费者接收到消息，执行任务
//            log.info("业务定时任务执行了" + new Date());
//            schedule.setStarted(1);
            scheduleMapper.updateStartedById(schedule.getId(), 1);

            ScheduleLog scheduleLog = new ScheduleLog()
                    .setScheduleId(schedule.getScheduleId())
                    .setMsgId(UUID.fastUUID().toString(true))
                    .setStartTime(LocalDateTime.now())
                    .setStatus(0);
            scheduleLogMapper.insert(scheduleLog);

            //发送消息到mq
            JobRequest jobRequest = new JobRequest()
                    .setScheduleId(schedule.getScheduleId())
                    .setMsgId(scheduleLog.getMsgId())
                    .setScheduleName(schedule.getScheduleName())
                    .setRedisTimeOut(schedule.getTimeout())
                    .setServer(schedule.getServer());
            jobSendProduce.sendJobMsg(jobRequest);
        };
    }

    /**
     * 判断任务是否可以运行，同步任务需要等上一个任务结束才能运行，异步不需要
     * @param schedule
     * @return
     */
    private boolean judgeRun(Schedule schedule) {
        Integer status = scheduleLogMapper.queryStatusByScheduleId(schedule.getScheduleId());
        return !schedule.getExecMode().equals("sync") || status != 0;
    }

    private Runnable newDaemonTask(ScheduledTaskRegistrar taskRegistrar) {
        return () -> {
            // 更新任务状态(超时的)
            checkTimeOut();

            //暂停、删除任务
            deleteTask();

            // 加载任务
            loadTask(taskRegistrar);
        };
    }

    private void checkTimeOut() {
        if (CollectionUtils.isEmpty(runTaskMap)) {
            return;
        }
        List<Schedule> scheduleList = this.scheduleMapper.selectList(new QueryWrapper<Schedule>()
                .in("schedule_id", runTaskMap.keySet())
                .eq("started", 1)
        );
        scheduleList.forEach(schedule -> {
            if (schedule.getTimeout() != null && schedule.getTimeout() > 0) {
                // 超时处理
                ScheduleLog scheduleLog = scheduleLogMapper.selectOne(new QueryWrapper<ScheduleLog>()
                        .eq("schedule_id", schedule.getScheduleId())
                        .eq("status", 0)
                        .orderByDesc("start_time")
                        .last("limit 1")
                );
                if (scheduleLog == null) {
                    return;
                }

                long between = LocalDateTimeUtil.between(scheduleLog.getStartTime(), LocalDateTime.now(), ChronoUnit.SECONDS);
                if (schedule.getTimeout() < between) {
                    scheduleLogMapper.updateStatusById(scheduleLog.getId(), 4);

                    scheduleMapper.updateStartedById(schedule.getId(), 0);
                }
            }
        });
    }

    private void deleteTask() {
        if (CollectionUtils.isEmpty(runTaskMap)) {
            return;
        }
        //查询禁用的任务
        List<Schedule> unEnabledList = scheduleMapper.selectList(new QueryWrapper<Schedule>().eq("enabled", 0));
        for (Schedule unEnabled : unEnabledList) {
            if (runTaskMap.containsKey(unEnabled.getScheduleId())) {
                ScheduledTask scheduledTask = runTaskMap.get(unEnabled.getScheduleId());
                scheduledTask.cancel();
                runTaskMap.remove(unEnabled.getScheduleId());
            }
        }

        //查询已删除的任务
        if (CollectionUtils.isEmpty(runTaskMap)) {
            return;
        }
        List<Schedule> runTaskList = scheduleMapper.selectList(new QueryWrapper<Schedule>().in("schedule_id", runTaskMap.keySet()).eq("enabled", 1));
        Set<String> scheduleIds = runTaskList.stream().map(Schedule::getScheduleId).collect(Collectors.toSet());
        runTaskMap.keySet().removeIf(key -> !scheduleIds.contains(key));
    }

    /**
     * 加载任务
     *
     * @param taskRegistrar
     */
    private void loadTask(ScheduledTaskRegistrar taskRegistrar) {
        //查询启用的任务，并且没有被加载
        List<Schedule> waitStartList = scheduleMapper.selectList(
                new QueryWrapper<Schedule>()
                        .ne("type", DAEMON)
                        .eq("enabled", 1)
                        .notIn(!CollectionUtils.isEmpty(runTaskMap), "schedule_id", runTaskMap.keySet())
        );

        for (Schedule waitStart : waitStartList) {
            ScheduledTask scheduledTask = taskRegistrar.scheduleTriggerTask(new TriggerTask(newTask(taskRegistrar, waitStart), getTrigger(waitStart)));
            runTaskMap.put(waitStart.getScheduleId(), scheduledTask);
        }

    }

    private static Trigger getTrigger(Schedule schedule) {
        return triggerContext -> new CronTrigger(schedule.getCron()).nextExecutionTime(triggerContext);
    }
}
